package consumer

import (
	"context"
	"encoding/json"
	"paper-pro/app/chat/jobs/batchinsert/internal/svc"
	kqtypes "paper-pro/app/chat/jobs/types"
	dialogmodel "paper-pro/app/chat/model/chatdialog"
	msgmodel "paper-pro/app/chat/model/chatmsg"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/zeromicro/go-zero/core/logx"
	"go.mongodb.org/mongo-driver/bson/primitive"
)

type BatchInsertConsumer struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewBatchInsertConsumer(ctx context.Context, svcCtx *svc.ServiceContext) *BatchInsertConsumer {
	return &BatchInsertConsumer{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

func (c *BatchInsertConsumer) Consume(ctx context.Context, msgs []*kafka.Message) error {
	myCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()

	chatMsgs := make([]*msgmodel.ChatMsg, 0, len(msgs))
	diaActiveMap := make(map[string]*dialogmodel.ChatDialogActiveInfo)
	for _, msg := range msgs {
		var kqMsg kqtypes.KqChatMsg
		err := json.Unmarshal(msg.Value, &kqMsg)
		if err != nil {
			c.Errorf("【BatchInsertConsumer】Consume: Unmarshal msg error: %v", err)
			continue
		}
		oDialogId, err := primitive.ObjectIDFromHex(kqMsg.DialogId)
		if err != nil {
			c.Errorf("【BatchInsertConsumer】Consume: oDialogId convert failed: DialogId %s", kqMsg.DialogId)
			continue
		}
		chatTime := time.Unix(kqMsg.ChatAt, 0)
		chatMsgs = append(chatMsgs, &msgmodel.ChatMsg{
			Content:   kqMsg.Content,
			Role:      kqMsg.Role,
			ChatAt:    chatTime,
			DialogID:  oDialogId,
			UserId:    kqMsg.UserId,
			RequestId: kqMsg.RequestId,
		})
		if activeInfo, exist := diaActiveMap[kqMsg.DialogId]; exist {
			if activeInfo.LastActiveAt.Before(chatTime) {
				activeInfo.LastActiveAt = chatTime
			}
		} else {
			diaActiveMap[kqMsg.DialogId] = &dialogmodel.ChatDialogActiveInfo{
				ID:           oDialogId,
				LastActiveAt: chatTime,
				UserId:       kqMsg.UserId,
			}
		}
	}

	_, err := c.svcCtx.ChatMsgModel.InsertMany(myCtx, chatMsgs)
	if err != nil {
		c.Errorf("【BatchInsertConsumer】Consume: ChatMsgModel InsertMany error: %v", err)
	}

	activeInfos := make([]*dialogmodel.ChatDialogActiveInfo, 0, len(diaActiveMap))
	for _, activeInfo := range diaActiveMap {
		activeInfos = append(activeInfos, activeInfo)
	}
	_, err = c.svcCtx.ChatDialogModel.BatchUpdateLastActive(myCtx, activeInfos)
	if err != nil {
		c.Errorf("【BatchInsertConsumer】Consume: ChatDialogModel BatchUpdateLastActive error: %v", err)
	}

	return nil

}
